package org.lqk.file.transfer.client.netty;

import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.lqk.file.transfer.CommandCallBack;
import org.lqk.file.transfer.netty.RequestFuture;
import org.lqk.file.transfer.protocol.*;
import org.lqk.file.transfer.util.SemaphoreReleaseOnlyOnce;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.Channel;

public class NettyRemotingClientHelper {
	private final ConcurrentHashMap<Integer, RequestFuture> requestTable;
	private Semaphore semaphoreAsync = new Semaphore(2048);

	private Semaphore semaphoreOneway = new Semaphore(2048);

	private static Logger log = LoggerFactory.getLogger(NettyRemotingClientHelper.class);
	public NettyRemotingClientHelper(ConcurrentHashMap<Integer, RequestFuture> requestTable){
		this.requestTable = requestTable;
	}
	/**
	 * 异步递送消息，并保证得到应答
	 * 
	 * @param channel
	 * @param request
	 */
	public void invokeAsync(final Channel channel, final RemotingCommand request, long timeoutMills,
			CommandCallBack callBack) throws InterruptedException {
		boolean acquire = semaphoreAsync.tryAcquire(timeoutMills, TimeUnit.MILLISECONDS);
		if (acquire) {
			SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
			final RequestFuture requestFuture = RequestFuture.createAsynFuture(request.getOpaque(), timeoutMills, true,
					once, channel, callBack);
			try {
				requestTable.put(request.getOpaque(), requestFuture);
				channel.writeAndFlush(request).addListener((channelFuture) -> {
					if (channelFuture.isSuccess()) {
						requestFuture.putRequest(request);
						return;
					} else {
						requestFuture.release();
						requestTable.remove(request.getOpaque());
					}
				});
			} catch (Exception e) {
				requestFuture.release();
				requestTable.remove(request.getOpaque());
				e.printStackTrace();
			}
		} else {
			// TODO 超时处理
		}
	}

	/**
	 * 同步递送消息
	 * 
	 * @param channel
	 * @param request
	 * @return
	 * @throws Exception
	 */
	public RemotingCommand invokeSync(final Channel channel, final RemotingCommand request, long timeoutMills)
			throws Exception {
		final RequestFuture requestFuture = RequestFuture.createSyncFuture(request.getOpaque(), timeoutMills);
		System.out.println("发送消息:" + request.getOpaque() + "消息内容:" + request.getContent());
		try {
			channel.writeAndFlush(request).addListener((channelFuture) -> {
				if (channelFuture.isSuccess()) {
					requestTable.put(request.getOpaque(), requestFuture);
					requestFuture.putRequest(request);
					requestFuture.setSendRequestOK(true);
					return;
				} else {
					requestFuture.setSendRequestOK(false);
					return;
				}
			});

			RemotingCommand response = null;
			if (requestFuture.isSendRequestOK()) {
				try {
					response = requestFuture.waitResponse(10 * 1000);
				} catch (Exception e) {
					e.printStackTrace();
				}
			} else {
				// TODO 将任务放入线程池
				response = new RemotingCommand(CmdCode.File_Task.getCode(), RpcType.File_Request.getCode(), true,RemotingCommand.autoIncrement());
			}

			return response;
		} finally {

		}
	}

	/**
	 * 单向递送消息
	 * 
	 * @param channel
	 * @param request
	 */
	public void invokeOneway(final Channel channel, RemotingCommand request, long timeoutMills)
			throws InterruptedException {
		boolean acquire = semaphoreOneway.tryAcquire(timeoutMills, TimeUnit.MILLISECONDS);
		if (acquire) {
			try {
				final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
				CommandBody body = request.getBody();
				if (body instanceof FileSegmentRequest) {
					FileSegmentRequest f = (FileSegmentRequest) body;
					// byte[] b = new byte[f.getBlockSize()];
					// int i = (int) (f.getPosition()/100);
					// int i = (int)(f.getPosition()/Constant.BLOCK_SIZE);
					// Arrays.fill(b, (byte) i);
					// System.out.println(f.getContent().length+"---"+Arrays.toString(f.getContent()));
					log.debug("position {},blocksize {},filename {},crc32 {}", f.getPosition(), f.getBlockSize(),
							f.getFileName(), f.getCrc32());

				}
				channel.writeAndFlush(request).addListener(channelFuture -> {
					once.release();
				});
			} catch (Exception e) {

			}
		}
	}
}
